package com.retoc.oawx.task;

import com.rabbitmq.client.*;
import com.retoc.oawx.domain.MessageEntity;
import com.retoc.oawx.domain.MessageRefEntity;
import com.retoc.oawx.exception.OAException;
import com.retoc.oawx.service.MessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

@Component
@Slf4j
public class MessageTask {

    @Autowired
    private ConnectionFactory factory;

    @Autowired
    private MessageService messageService;

    public void send(String topic, MessageEntity message) {
        String id = messageService.insertMessage(message);
        try(Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();)
        {
            channel.queueDeclare(topic,true,false,false,null);
            HashMap map=new HashMap<>();
            map.put("messageId",id);
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(map).build();
            channel.basicPublish("",topic,properties,message.getMsg().getBytes(StandardCharsets.UTF_8));
            log.debug("消息发送成功");
        } catch (Exception e) {
            log.error("执行异常", e);
            throw new OAException("向MQ发送消息失败");
        }
    }

    @Async
    public void sendAsync(String topic, MessageEntity message) {
        send(topic,message);
    }

    public int received(String topic) {
        int i=0;
        try(Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();)
        {
            channel.queueDeclare(topic,true,false,false,null);
            while (true) {
                GetResponse response = channel.basicGet(topic, false);
                if (response != null) {
                    AMQP.BasicProperties props = response.getProps();
                    Map<String, Object> map = props.getHeaders();
                    String messageId = map.get("messageId").toString();
                    byte[] body = response.getBody();
                    String message = new String(body);
                    log.debug("从RabbitMQ接收消息："+message);
                    MessageRefEntity entity = new MessageRefEntity();
                    entity.setMessageId(messageId);
                    entity.setReceiverId(Integer.parseInt(topic));
                    entity.setReadFlag(false);
                    entity.setLastFlag(true);
                    messageService.insertRef(entity);
                    long deliveryTag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag, false);
                    i++;
                }
                else {
                    break;
                }
            }
        }
        catch (Exception e){
            log.error("执行异常", e);
            throw new OAException("接收消息失败");
        }
        return i;
    }

    @Async
    public int receivedAsync(String topic) {
        return received(topic);
    }

    public void deleteQueue(String topic) {
        try(Connection connection=factory.newConnection();
            Channel channel=connection.createChannel();)
        {
            channel.queueDelete(topic);
        }
        catch (Exception e){
            log.error("删除消息异常", e);
            throw new OAException("删除消息失败");
        }
    }

    @Async
    public void deleteQueueAsync(String topic) {
        deleteQueue(topic);
    }
}
